package com.jiaoxiaobo.consumer;

import com.jiaoxiaobo.RabbitMQUtil;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.HashMap;

public class StreamConsumer {

    private static final String QUEUE_NAME = "hello-stream-queue";

    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.basicQos(100);

        //声明stream队列
        HashMap<String, Object> map = new HashMap<>();
        map.put("x-queue-type","stream");
        map.put("x-max-length-bytes",20_000_000_000L);
        map.put("x-stream-max-segment-size-bytes",100_000_000L);

        channel.queueDeclare(QUEUE_NAME,true,false,false,map);

        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                System.out.println("========================");
                String routingKey = envelope.getRoutingKey();
                System.out.println("routingKey >"+routingKey);
                String contentType = properties.getContentType();
                System.out.println("contentType >"+contentType);
                long deliveryTag = envelope.getDeliveryTag();
                System.out.println("deliveryTag >"+deliveryTag);
                System.out.println("content:"+new String(body,"UTF-8"));
                System.out.println("messageId:"+properties.getMessageId());
                properties.getHeaders().forEach((key,value)-> System.out.println("key: "+key +"; value: "+value));
                channel.basicAck(deliveryTag,false);
            }
        };

        //设置消费模式
        //first:从日志队列中第一个可消费的消息开始消费
        //last:消费消息日志中的最后一个消息
        //next:相当于不指定offset，消费不到消息
        //offset：一个数字型的偏移量
        //Timestamp:一个代表时间的Date类型变量，表示从这个时间点开始消费
        HashMap<String, Object> consumerParam = new HashMap<>();
        consumerParam.put("x-stream-offset","first");
        channel.basicConsume(QUEUE_NAME,false,consumerParam,consumer);
    }

}
